package com.czk.dao;

import com.czk.constant.Constants;
import com.czk.utils.HbaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 1 发布微博
 * 2 删除微博
 * 3 关注用户
 * 4 取关用户
 * 5 获取微博用户详情
 * 6 获取用户的初始化页面
 *
 * @Author:ChenZhangKun
 * @Date: 2021/12/3 14:53
 */
public class HBaseDao {
    /**
     * 发布微博
     *
     * @param uid
     * @param content
     */
    public static void publishWeiBo(String uid, String content) throws IOException {
        Connection connection = null;
        Table table = null;
        Table relationTable = null;
        try {
            // 获取connection
            connection = ConnectionFactory.createConnection(Constants.configuration);
            // 操作微博内容
            table = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
            // 获取当前时间戳
            long ts = System.currentTimeMillis();
            // 定义rowKey
            String rowKey = uid + "_" + ts;
            // 创建put对象
            Put contPut = new Put(Bytes.toBytes(rowKey));
            // 赋值 Bytes.toBytes("content") 列名称  content->内容
            contPut.addColumn(Bytes.toBytes(Constants.CONTENT_INFO_CF), Bytes.toBytes("content"), Bytes.toBytes(content));
            // 插入数据
            table.put(contPut);

            // 操作微博收件箱表
            // 1 获取用户关系表对象
            relationTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
            // 获取粉丝列族数据
            Get get = new Get(Bytes.toBytes(uid));
            // 指定粉丝列族
            get.addFamily(Bytes.toBytes(Constants.RELATION_INFO_CF2));
            Result result = relationTable.get(get);
            List<Put> inboxPuts = new ArrayList<>();
            // 创建集合，用户存放微博内容表的Put对象
            for (Cell cell : result.rawCells()) {
                // 构建微博收件箱的put对象
                Put inboxPut = new Put(CellUtil.cloneQualifier(cell));
                // 给收件箱的put对象赋值->最新发布微博的rowKey
                inboxPut.addColumn(Bytes.toBytes(Constants.INBOX_INFO_CF), Bytes.toBytes(uid), Bytes.toBytes(rowKey));
                // 加入集合
                inboxPuts.add(inboxPut);
            }
            if (inboxPuts.size() > 0) {
                // 大于0，有粉丝
                Table inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
                // 插入收件箱表
                inboxTable.put(inboxPuts);
                // 关闭资源
                inboxTable.close();
            }
        } finally {
            // 关闭资源
            HbaseUtil.close(connection, table, relationTable);
        }
    }

    /**
     * 删除微博
     *
     * @param rowKey
     */
    public static void deleteWeiBo(String rowKey) {
        // 删除微博表
        // 删除邮件表里面的微博
    }

    /**
     * 添加关注
     *
     * @param uid
     * @param attends
     */
    public static void addAttends(String uid, String... attends) {
        // 校验是否添加了带关注的人
        if (attends.length <= 0) {
            System.out.println("请选择需要关注的人");
            return;
        }
        Connection connection = null;
        Table relationTable = null;
        Table inboxTable = null;
        Table contTable = null;
        try {
            // 获取链接
            connection = ConnectionFactory.createConnection(Constants.configuration);
            /// 获取用户关系表
            relationTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
            // 创建集合，用户存储用户关系表的put对象
            List<Put> relaPuts = new ArrayList<>();
            // 创建操作者的put对象
            Put uidPut = new Put(Bytes.toBytes(uid));
            // 循环添加被关注者的Put对象
            for (String attend : attends) {
                // 给操作者的put对象赋值
                uidPut.addColumn(Bytes.toBytes(Constants.RELATION_INFO_CF1), Bytes.toBytes(attend), Bytes.toBytes(attend));
                // 创建备注着的put对象
                Put attendPut = new Put(Bytes.toBytes(attend));
                attendPut.addColumn(Bytes.toBytes(Constants.RELATION_INFO_CF2), Bytes.toBytes(uid), Bytes.toBytes(uid));
                // 加入集合
                relaPuts.add(attendPut);
            }
            relaPuts.add(uidPut);
            // 执行插入
            relationTable.put(relaPuts);

            // 微博内容表操作
            inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
            Put inboxInput = new Put(Bytes.toBytes(uid));
            // 内容表对象
            contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
            for (String attend : attends) {
                // 获取被关注者的微博
                Scan scan = new Scan(Bytes.toBytes(attend + "_"), Bytes.toBytes(attend + "|"));
                // 拿到内容
                ResultScanner resultScanner = contTable.getScanner(scan);
                // 定义时间戳
                long ts = System.currentTimeMillis();
                for (Result result : resultScanner) {
                    // 拿到微博内容rowkey 加入收件箱表
                    // hbase添加数据是增加个时间戳
                    inboxInput.addColumn(Bytes.toBytes(Constants.INBOX_INFO_CF), Bytes.toBytes(attend), ts++, result.getRow());
                }
            }
            if (!inboxInput.isEmpty()) {
                inboxTable.put(inboxInput);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            HbaseUtil.close(connection, relationTable, inboxTable, contTable);
        }

    }

    /**
     * 取消关注
     *
     * @param uid
     * @param attends
     */
    public static void deleteAttends(String uid, String... attends) {
        Connection connection = null;
        Table relationTable = null;
        Table inboxTable = null;

        try {
            connection = ConnectionFactory.createConnection(Constants.configuration);
            // 拿到关系表
            relationTable = connection.getTable(TableName.valueOf(Constants.RELATION_TABLE));
            // 操作者的delete对象
            Delete uidDelete = new Delete(Bytes.toBytes(uid));
            List<Delete> list = new ArrayList<>();
            for (String attend : attends) {
                // 创建delete对象
                uidDelete.addColumns(Bytes.toBytes(Constants.RELATION_INFO_CF1), Bytes.toBytes(attend));
                // 被取关者
                Delete delete = new Delete(Bytes.toBytes(attend));
                delete.addColumns(Bytes.toBytes(Constants.RELATION_INFO_CF2), Bytes.toBytes(uid));
                list.add(delete);
            }
            list.add(uidDelete);
            // 删除
            relationTable.delete(list);

            // 获取邮箱表
            inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));

            Delete delete = new Delete(Bytes.toBytes(uid));
            for (String attend : attends) {
                delete.addColumns(Bytes.toBytes(Constants.INBOX_INFO_CF), Bytes.toBytes(attend));
            }
            inboxTable.delete(delete);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            HbaseUtil.close(connection, relationTable, inboxTable);

        }
    }

    /**
     * 拿到关注人的微博
     *
     * @param uid
     */
    public static void getInit(String uid) {
        Connection connection = null;
        Table inboxTable = null;
        Table contTable = null;
        try {
            connection = ConnectionFactory.createConnection(Constants.configuration);
            // 拿到邮箱表
            inboxTable = connection.getTable(TableName.valueOf(Constants.INBOX_TABLE));
            // get对象
            contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
            Get inboxGet = new Get(Bytes.toBytes(uid));
            Result result = inboxTable.get(inboxGet);
            // 获取数据
            for (Cell cell : result.rawCells()) {
                // 构建微博内容表的
                Get get = new Get(CellUtil.cloneValue(cell));
                // 获取对象
                Result contentResult = contTable.get(get);
                // 获取内容
                for (Cell contCell : contentResult.rawCells()) {
                    System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(contCell)) +
                            ",CF:" + Bytes.toString(CellUtil.cloneFamily(contCell)) +
                            ",CN:" + Bytes.toString(CellUtil.cloneQualifier(contCell)) +
                            ",Value" + Bytes.toString(CellUtil.cloneValue(contCell)));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            HbaseUtil.close(connection, inboxTable, contTable);
        }
    }

    /**
     * 获取某个人的微博详情
     *
     * @param uid
     */
    public static void getWeiBo(String uid) {
        Connection connection = null;
        Table contTable = null;
        try {
            connection = ConnectionFactory.createConnection(Constants.configuration);
            // 获取微博内容表
            contTable = connection.getTable(TableName.valueOf(Constants.CONTENT_TABLE));
            //
            Scan scan = new Scan();
            // 设置过滤器
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(uid + "_"));
            scan.setFilter(rowFilter);
            ResultScanner resultScanner = contTable.getScanner(scan);
            for (Result result : resultScanner) {
                for (Cell contCell : result.rawCells()) {
                    System.out.println("RK:" + Bytes.toString(CellUtil.cloneRow(contCell)) +
                            ",CF:" + Bytes.toString(CellUtil.cloneFamily(contCell)) +
                            ",CN:" + Bytes.toString(CellUtil.cloneQualifier(contCell)) +
                            ",Value" + Bytes.toString(CellUtil.cloneValue(contCell)));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            HbaseUtil.close(connection, contTable);
        }
    }
}
